/*
* Copyright (C) Microsoft. All rights reserved.  
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not 
* use this file except in compliance with the License.  You may obtain a copy 
* of the License at http://www.apache.org/licenses/LICENSE-2.0  
* 
* THIS CODE IS PROVIDED *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY IMPLIED 
* WARRANTIES OR CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE, 
* MERCHANTABLITY OR NON-INFRINGEMENT. 
*
* See the Apache License, Version 2.0 for specific language governing 
* permissions and limitations under the License.
*
*/
#include "pch.h"

#include <baja/error.hpp>
#include <baja/assert.hpp>
#include <baja/db/engines/jet/db_jet_error.h>
#include <baja/db/engines/jet/db_jet_engine.h>
#include <baja/db/db_result.h>

#pragma comment(lib, "esent.lib")

using namespace baja::storage;

db_jet_engine::db_jet_engine() :
        _instance(JET_instanceNil),
        _sessionId(JET_sesidNil),
        _dbID(JET_dbidNil)
{
    _updateRowBinding = std::bind(&db_jet_engine::updateRow,
                                  this,
                                  std::placeholders::_1,
                                  std::placeholders::_2,
                                  std::placeholders::_3);
    _readRowBinding = std::bind(&db_jet_engine::readRow,
                                  this,
                                  std::placeholders::_1,
                                  std::placeholders::_2,
                                  std::placeholders::_3);
    _deleteRowBinding = std::bind(&db_jet_engine::deleteRow,
                                  this,
                                  std::placeholders::_1,
                                  std::placeholders::_2,
                                  std::placeholders::_3);
}

std::shared_ptr<iengine> db_jet_engine::create()
{
    return std::make_shared<db_jet_engine>();
}

db_jet_engine::~db_jet_engine()
{
    shutdown();
}

void db_jet_engine::shutdown()
{
    if (_dbID != JET_dbidNil)
    {
        JetCloseDatabase(_sessionId, _dbID, 0 /* reserved */);
        _dbID = JET_dbidNil;
    }
    if (_sessionId != JET_sesidNil)
    {
        JetEndSession(_sessionId, 0 /* reserved */);
        _sessionId = JET_sesidNil;
    }
    if (_instance != JET_instanceNil)
    {
        JetTerm2(_instance, JET_bitTermComplete);
        _instance = JET_instanceNil;
    }
}

void db_jet_engine::setPageSize(unsigned int size)
{
    jet_call(
        ::JetSetSystemParameter(
            &_instance, 
            _sessionId, 
            JET_paramDatabasePageSize, 
            size, 
            nullptr
            ));
}

void db_jet_engine::setTempPath(std::wstring tempPath)
{
    jet_call(
        ::JetSetSystemParameter(
            &_instance, 
            0, 
            JET_paramTempPath, 
            0, 
            tempPath.c_str()
            ));
}

void db_jet_engine::setLogFilePath(std::wstring logFilePath)
{
    jet_call(
        ::JetSetSystemParameter(
            &_instance, 
            0, 
            JET_paramLogFilePath, 
            0, 
            logFilePath.c_str()
            ));
}

void db_jet_engine::setSystemPath(std::wstring systemPath)
{
    jet_call(
        ::JetSetSystemParameter(
            &_instance, 
            0, 
            JET_paramSystemPath, 
            0, 
            systemPath.c_str()
            )); 
}

void db_jet_engine::createInstance(std::wstring workingFolderName)
{
    jet_call(
        ::JetCreateInstance2(
            &_instance,
            L"{0A9A6617-8AE9-4c5e-AF28-01D5D4820C23}",
            nullptr,
            0));

    setTempPath(workingFolderName);
    setLogFilePath(workingFolderName);
    setSystemPath(workingFolderName);

    jet_call(
        ::JetInit3(
            &_instance,
            nullptr,
            0
            ));
}

void db_jet_engine::beginSession()
{
    jet_call(
        ::JetBeginSession(
            _instance,    // zero as ESE instance
            &_sessionId,  // return value
            0,            // reserved
            0             // reserved
            ));
}

void db_jet_engine::createDatabase(std::wstring fileName)
{
    jet_call(
        ::JetCreateDatabase2(
            _sessionId,   // the session identifier
            fileName.c_str(),    // file name
            0,            // reserved
            &_dbID,       // return value
            0             // zero flag - just create the database
            ));
}

void db_jet_engine::openDatabase(std::wstring fileName)
{
    JET_ERR err = JET_errSuccess;
    err = ::JetAttachDatabase2(
            _sessionId,
            fileName.c_str(),
            0,
            0);

    if (err != JET_errSuccess)
    {
        if (err == JET_errFileNotFound)
        {
            createDatabase(fileName);
        }
        else
        {
            BAJA_THROW(jet_exception("Failed to attach database!", err));
        }
    }
    else
    {
        jet_call(
            ::JetOpenDatabaseW(
                _sessionId,
                fileName.c_str(),
                nullptr,
                &_dbID,
                0 // not exclusive, and not read-only...
                ));
    }
}

void db_jet_engine::openTable(std::shared_ptr<itable> tableDefinition)
{
    BAJA_LOCK(_mutex);

    JET_TABLEID tableId;

    JET_ERR err = JET_errSuccess;
    err = JetOpenTable(
            _sessionId,
            _dbID,
            tableDefinition->getName().c_str(),
            nullptr,
            0,
            0,
            &tableId);

    if (err != JET_errSuccess)
    {
        if (err == JET_errObjectNotFound)
        {
            createTable(tableDefinition);
        }
        else
        {
            BAJA_THROW(jet_exception("Failed to open table!", err));
        }
    }
    else
    {
        tableDefinition->setId(tableId);
        // we need to get the column ids...
        std::shared_ptr<std::vector<std::shared_ptr<icolumn>>> columns = tableDefinition->getColumns();
        std::vector<std::shared_ptr<icolumn>>::iterator columnIt;
        for (columnIt = columns->begin(); columnIt < columns->end(); ++columnIt)
        {
            std::shared_ptr<icolumn> column = *columnIt;

            JET_COLUMNDEF result;
            unsigned long resultSize = sizeof(JET_COLUMNDEF);

            jet_call(
                JetGetColumnInfo(
                    _sessionId,
                    _dbID,
                    tableDefinition->getName().c_str(),
                    column->getName().c_str(),
                    &result, 
                    resultSize,
                    JET_ColInfo));

            // We assume we only need the columnid since we know the rest.
            column->setId(result.columnid);
        }
    }

}

void db_jet_engine::createTable(std::shared_ptr<itable> tableDefinition)
{
    BAJA_LOCK(_mutex);

    strings_buffer tempStrings;

    // Build an array of column definitions...
    std::vector<JET_COLUMNCREATE> eseColumns;
    std::shared_ptr<std::vector<std::shared_ptr<icolumn>>> columns = tableDefinition->getColumns();
    std::vector<std::shared_ptr<icolumn>>::iterator itColumn;
    for (itColumn = columns->begin() ; itColumn < columns->end() ; ++itColumn)
    {
        std::shared_ptr<icolumn> column = *itColumn;

        JET_COLUMNCREATE eseColumnDefinition;
        memset(&eseColumnDefinition, 0, sizeof(JET_COLUMNCREATE));
        eseColumnDefinition.cbStruct = sizeof(JET_COLUMNCREATE);
        std::wstring* tempColumnName = new std::wstring(column->getName());      // We need to create it on the heap to make sure the buffer 
        tempStrings.strings.push_back(tempColumnName);               // does not move until JetCreateTableColumnIndex() is complete.
        eseColumnDefinition.szColumnName = const_cast<wchar_t*>(tempColumnName->c_str());
        switch (column->getType())
        {
        case type_int16:
            eseColumnDefinition.coltyp = JET_coltypShort;
            eseColumnDefinition.cbMax = sizeof(unsigned short);
            break;
        case type_int32:
            eseColumnDefinition.coltyp = JET_coltypLong;
            eseColumnDefinition.cbMax = sizeof(unsigned int);
            break;
        case type_int64:
            eseColumnDefinition.coltyp = JET_coltypLongLong;
            eseColumnDefinition.cbMax = sizeof(unsigned int);
            break;
        case type_float32:
            eseColumnDefinition.coltyp = JET_coltypIEEESingle;
            eseColumnDefinition.cbMax = sizeof(float);
            break;
        case type_string:
            BAJA_ASSERT(column->getMaxBufferSizeInBytes() <= 255, L"type_string cannot have more than 255 characters");
            eseColumnDefinition.coltyp = JET_coltypText;
            eseColumnDefinition.cbMax = column->getMaxBufferSizeInBytes();
            eseColumnDefinition.cp = column->getCodePage();
            break;
        case type_string_long: // greater than 255 ascii, or 127 unicode
            eseColumnDefinition.coltyp = JET_coltypLongText;
            eseColumnDefinition.cbMax = column->getMaxBufferSizeInBytes();
            eseColumnDefinition.cp = column->getCodePage();
            break;
        case type_binaryBlob:
            eseColumnDefinition.coltyp = JET_coltypLongBinary;
            eseColumnDefinition.cbMax = column->getMaxBufferSizeInBytes();
            eseColumnDefinition.grbit = JET_bitColumnTagged;
            break;
        default:
            BAJA_ASSERT(false, L"Unsupported data type!");
        }

        // column options
        if (column->getFixed())
        {
            eseColumnDefinition.grbit |= JET_bitColumnFixed;
        }
        if (column->getAutoIncrement())
        {
            eseColumnDefinition.grbit |= JET_bitColumnAutoincrement;
        }
        eseColumnDefinition.err = JET_errSuccess;

        eseColumns.push_back(eseColumnDefinition);
    }

    // Build an array of column definitions...
    std::vector<JET_INDEXCREATE3> eseIndices;
    std::shared_ptr<std::vector<std::shared_ptr<iindex>>> indices = tableDefinition->getIndices();
    std::vector<std::shared_ptr<iindex>>::iterator itIndex;
    for (itIndex = indices->begin() ; itIndex < indices->end() ; ++itIndex)
    {
        std::shared_ptr<iindex> index = *itIndex;

        JET_INDEXCREATE3 eseIndexDefinition;
        memset(&eseIndexDefinition, 0, sizeof(JET_INDEXCREATE3));
        eseIndexDefinition.cbStruct = sizeof(JET_INDEXCREATE3);
        std::wstring* tempIndexName = new std::wstring(index->getName());      // We need to create it on the heap to make sure the buffer 
        tempStrings.strings.push_back(tempIndexName);                          // does not move until JetCreateTableColumnIndex() is complete.
        eseIndexDefinition.szIndexName = const_cast<wchar_t*>(tempIndexName->c_str());
        if (index->getPrimary())
        {
            eseIndexDefinition.grbit |= JET_bitIndexPrimary;
        }
        if (index->getUnique())
        {
            eseIndexDefinition.grbit |= JET_bitIndexUnique;
        }

        // Calculate the buffer...
        std::wstring* tempKey = new std::wstring();
        std::shared_ptr<std::vector<std::shared_ptr<iindex_column>>> indexColumns = index->getIndexColumns();
        std::vector<std::shared_ptr<iindex_column>>::iterator itIndexColumn;
        for (itIndexColumn = indexColumns->begin(); itIndexColumn < indexColumns->end(); ++itIndexColumn)
        {
            std::shared_ptr<iindex_column> indexColumn = *itIndexColumn;
            std::shared_ptr<icolumn> column = indexColumn->getColumn();

            (*tempKey) += indexColumn->getAscending() ? L"+" : L"-";
            (*tempKey) += std::wstring(column->getName().c_str(), column->getName().size() + 1); // +/- and \0
        }
        tempStrings.strings.push_back(tempKey);
        eseIndexDefinition.szKey = const_cast<wchar_t*>(tempKey->c_str());
        eseIndexDefinition.cbKey = ( static_cast<unsigned long>(tempKey->size()) + 1 ) * sizeof(wchar_t);
        eseIndexDefinition.err = JET_errSuccess;

        eseIndices.push_back(eseIndexDefinition);
    }

    JET_TABLECREATE4 eseTableDefinition;
    memset(&eseTableDefinition, 0, sizeof(JET_TABLECREATE4));
    eseTableDefinition.cbStruct = sizeof(JET_TABLECREATE4);
    std::wstring* tempTableName = new std::wstring(tableDefinition->getName());
    tempStrings.strings.push_back(tempTableName);
    eseTableDefinition.szTableName = const_cast<wchar_t*>(tempTableName->c_str());
    eseTableDefinition.rgcolumncreate = eseColumns.size() ? &(eseColumns[0]) : nullptr;
    eseTableDefinition.cColumns = static_cast<unsigned long>(eseColumns.size());
    eseTableDefinition.rgindexcreate = eseIndices.size() ? &(eseIndices[0]) : nullptr;
    eseTableDefinition.cIndexes = static_cast<unsigned long>(eseIndices.size());
    eseTableDefinition.grbit = 0;

    JET_ERR err = JET_errSuccess;
    err = JetCreateTableColumnIndex4(
        _sessionId, 
        _dbID,
        &eseTableDefinition);

    if (JET_errSuccess != err)
    {
        if (err != JET_errTableDuplicate)
        {
            // We could read the .err attribute of each column/index if we need to know which one exactly failed.
            BAJA_THROW(jet_exception("Failed to create table!", err));
        }
    }

    // Read and store the new ids...
    tableDefinition->setId(eseTableDefinition.tableid);

    unsigned int i = 0;
    for (itColumn = columns->begin() ; itColumn < columns->end() ; ++i, ++itColumn)
    {
        std::shared_ptr<icolumn> column = *itColumn;
        column->setId(eseColumns[i].columnid);
    }
}

void db_jet_engine::insert(std::shared_ptr<itable> table,
                           std::shared_ptr<std::vector<std::shared_ptr<icell>>> row)
{
    BAJA_LOCK(_mutex);

    writeRow(table, row, insertRecord);
}

void db_jet_engine::update(std::shared_ptr<itable> table,
                           std::shared_ptr<iindex> index,
                           std::shared_ptr<std::vector<std::shared_ptr<icell>>> criteriaRow,
                           baja::uint32 criteriaCount,
                           bool assumeSingleMatch)
{
    BAJA_LOCK(_mutex);

    forEachMatch(table, index, criteriaRow, criteriaCount, assumeSingleMatch, _updateRowBinding, nullptr);
}

void db_jet_engine::read(std::shared_ptr<itable> table,
                         std::shared_ptr<iindex> index,
                         std::shared_ptr<std::vector<std::shared_ptr<icell>>> criteriaRow,
                         baja::uint32 criteriaCount,
                         bool assumeSingleMatch,
                         std::shared_ptr<iresult> result)
{
    BAJA_LOCK(_mutex);

    forEachMatch(table, index, criteriaRow, criteriaCount, assumeSingleMatch, _readRowBinding, result);
}

void db_jet_engine::dispose(
    std::shared_ptr<itable> table,
    std::shared_ptr<iindex> index,
    std::shared_ptr<std::vector<std::shared_ptr<icell>>> criteriaRow,
    b::uint32 criteriaCount,
    bool assumeSingleMatch)
{
    BAJA_LOCK(_mutex);

    forEachMatch(table, index, criteriaRow, criteriaCount, assumeSingleMatch, _deleteRowBinding, nullptr);
}

void db_jet_engine::readAll(std::shared_ptr<itable> table, std::shared_ptr<iindex> index, std::shared_ptr<iresult> result)
{
    BAJA_LOCK(_mutex);

    forAll(table, index, _readRowBinding, result);
}

void db_jet_engine::disposeAll(std::shared_ptr<itable> table, std::shared_ptr<iindex> index)
{
    BAJA_LOCK(_mutex);

    forAll(table, index, _deleteRowBinding, nullptr);
}

void db_jet_engine::writeRow(std::shared_ptr<itable> table,
                             std::shared_ptr<std::vector<std::shared_ptr<icell>>> inputRow,
                             operation_type operationType)
{
    BAJA_ASSERT( (operationType == insertRecord) || (operationType == updateRecord), L"db_jet_engine::write() accept only update/insert operations" );

    jet_call(
        ::JetBeginTransaction3(
            _sessionId,
            0,
            0));

    try
    {
        jet_call(
            ::JetPrepareUpdate(
                _sessionId,
                table->getId(),
                operationType == insertRecord ? JET_prepInsert : JET_prepReplace
                ));

        b::uint32 blobSize = 0;

        // Build the columns array...
        std::vector<JET_SETCOLUMN> cells;
        std::vector<std::shared_ptr<icell>>::iterator cellIt;
        for (cellIt = inputRow->begin() ; cellIt < inputRow->end() ; ++cellIt)
        {
            JET_SETCOLUMN eseCell = { 0 }; 
            JET_SETCOLUMN eseSizeCell = { 0 };

            eseCell.columnid = (*cellIt)->getColumnId(); 
            eseCell.pvData = (*cellIt)->getBuffer(); 
            eseCell.cbData = static_cast<unsigned long>((*cellIt)->getSize());

            if ( (*cellIt)->getColumn()->getFixed() == false )
            {
                eseCell.grbit = JET_bitSetSeparateLV;

                if ( (*cellIt)->getAppendOnUpdate() == true )
                {
                    eseCell.grbit = JET_bitSetAppendLV; 
                }
                else
                {
                    eseCell.grbit = JET_bitSetOverwriteLV; 

                    blobSize = static_cast<b::uint32>((*cellIt)->getSize());;

                    eseSizeCell.columnid = (*cellIt)->getColumnId(); 
                    eseSizeCell.pvData = &blobSize;
                    eseSizeCell.cbData = blobSize;
                    eseSizeCell.grbit = JET_bitSetSizeLV;
                    eseSizeCell.itagSequence = 1;
                    eseSizeCell.err = JET_errSuccess;

                    cells.push_back(eseSizeCell);
                }
                // tagged columns == multi-valued columents.
                // itagSequence = 0 : append a new value.
                // itagSequence = n (where n > 0), set value #n.
                eseCell.itagSequence = 1;
            }
            eseCell.err = JET_errSuccess; 

            cells.push_back(eseCell);
        }

        jet_call(
            ::JetSetColumns(
                _sessionId,
                table->getId(),
                &(cells[0]), static_cast<unsigned long>(cells.size())
                )); 

        jet_call(
            ::JetUpdate2(
                _sessionId,
                table->getId(),
                0,
                0,
                0,
                0));

        jet_call(
            ::JetCommitTransaction(
                _sessionId,
                0)); 
    }
    catch(jet_exception& e)
    {
        ::JetRollback(_sessionId, 0); 
        throw e;
    }
}

void db_jet_engine::updateRow(std::shared_ptr<itable> table,
                              std::shared_ptr<std::vector<std::shared_ptr<icell>>> row,
                              std::shared_ptr<iresult>)
{
    writeRow(table, row, updateRecord);
}

void db_jet_engine::readRow(std::shared_ptr<itable> table,
                            std::shared_ptr<std::vector<std::shared_ptr<icell>>> dummyRow,
                            std::shared_ptr<iresult> result)
{
    JET_ERR err = JET_errSuccess;

    unsigned int blobSize = 0;

    std::shared_ptr<iresult_row> row = createRow();

    std::shared_ptr<std::vector<std::shared_ptr<icolumn>>> columns = result->getColumns(); // table->getColumns();
    std::vector<std::shared_ptr<icolumn>>::iterator itColumn;
    for (itColumn = columns->begin() ; itColumn < columns->end() ; ++itColumn)
    {
        std::shared_ptr<icolumn> column = *itColumn;
        unsigned long readBytes = 0;

        switch (column->getType())
        {
        case type_int16:
            {
                unsigned short value;
                jet_call(
                    ::JetRetrieveColumn(_sessionId,
                        table->getId(),
                        column->getId(),
                        &value,
                        sizeof(value),
                        &readBytes,
                        0,
                        0
                        ),
                        "Failed to read type_int16 value from table! JetRetrieveColumn()");

                BAJA_ASSERT(readBytes == sizeof(value), L"DB inconsistency with memory objects. Has a recent schema change took place?");

                if (column->getIsBlobSize() == true)
                {
                    blobSize = value;
                }
                std::shared_ptr<icolumn_typed<unsigned short>> typedColumn = std::dynamic_pointer_cast<icolumn_typed<unsigned short>>( column );
                std::shared_ptr<icell_typed<unsigned short>> cell = typedColumn->createCell(value);
                row->addCell(cell);
            }
            break;
        case type_int32:
            {
                int value;
                jet_call(
                    ::JetRetrieveColumn(
                        _sessionId,
                        table->getId(),
                        column->getId(),
                        &value,
                        sizeof(value),
                        &readBytes,
                        0,
                        0),
                        "Failed to read type_int32 value from table! JetRetrieveColumn()");

                BAJA_ASSERT(readBytes == sizeof(value), L"DB inconsistency with memory objects. Has a recent schema change took place?");

                if (column->getIsBlobSize() == true)
                {
                    blobSize = value;
                }
                std::shared_ptr<icolumn_typed<unsigned int>> typedColumn = std::dynamic_pointer_cast<icolumn_typed<unsigned int>>( column );
                std::shared_ptr<icell_typed<unsigned int>> cell = typedColumn->createCell(value);
                row->addCell(cell);
            }
            break;
        case type_int64:
            {
                __int64 value;
                jet_call(
                    ::JetRetrieveColumn(_sessionId,
                        table->getId(),
                        column->getId(),
                        &value,
                        sizeof(value),
                        &readBytes,
                        0,
                        0),
                        "Failed to read type_int64 value from table! JetRetrieveColumn()");

                BAJA_ASSERT(readBytes == sizeof(value), L"DB inconsistency with memory objects. Has a recent schema change took place?");

                std::shared_ptr<icolumn_typed<unsigned __int64>> typedColumn = std::dynamic_pointer_cast<icolumn_typed<unsigned __int64>>( column );
                std::shared_ptr<icell_typed<unsigned __int64>> cell = typedColumn->createCell(value);
                row->addCell(cell);
            }
            break;

        case type_float32:
            {
                float value;
                jet_call(
                    ::JetRetrieveColumn(_sessionId,
                        table->getId(),
                        column->getId(),
                        &value,
                        sizeof(value),
                        &readBytes,
                        0,
                        0),
                        "Failed to read type_float32 value from table! JetRetrieveColumn()");

                BAJA_ASSERT(readBytes == sizeof(value), L"DB inconsistency with memory objects. Has a recent schema change took place?");

                std::shared_ptr<icolumn_typed<float>> typedColumn = std::dynamic_pointer_cast<icolumn_typed<float>>( column );
                std::shared_ptr<icell_typed<float>> cell = typedColumn->createCell(value);
                row->addCell(cell);
            }
            break;
        case type_string:
        case type_string_long:
            {
                baja::uint32 sizeInBytes = column->getMaxBufferSizeInBytes();

                string_buffer stringBuffer(sizeInBytes + sizeof(wchar_t));

                jet_call(
                    ::JetRetrieveColumn(_sessionId,
                        table->getId(),
                        column->getId(),
                        stringBuffer.buffer,
                        sizeInBytes,
                        &readBytes,
                        0,
                        0
                        ),
                        "Failed to read type_string/type_string_long value from table! JetRetrieveColumn()");

                if (column->getType() == type_string)
                {
                    // type_string is a fixed length type, so strings will always be appended with spaces.
                    BAJA_ASSERT(readBytes == sizeInBytes, L"DB inconsistency with memory objects. Has a recent schema change took place?");

                    stringBuffer.rightTrimSpaces();
                }
                else
                {
                    stringBuffer.buffer[readBytes / sizeof(wchar_t)] = L'\0';
                }

                std::shared_ptr<icolumn_typed<std::wstring>> typedColumn = std::dynamic_pointer_cast<icolumn_typed<std::wstring>>( column );
                std::shared_ptr<icell_typed<std::wstring>> cell = typedColumn->createCell(std::wstring(stringBuffer.buffer));
                row->addCell(cell);
            }
            break;
        case type_binaryBlob:
            {
                binary_blob blob(blobSize);

                if (blobSize != 0)
                {
                    jet_call(
                        ::JetRetrieveColumn(_sessionId,
                        table->getId(),
                        column->getId(),
                        blob._buffer,
                        static_cast<unsigned long>(blob._size),
                        &readBytes,
                        0,
                        0
                        ),
                        "Failed to read type_blob value from table! JetRetrieveColumn()");

                    BAJA_ASSERT(readBytes == blob._size, L"DB inconsistency with memory objects. Has a recent schema change took place?");
                }

                std::shared_ptr<icolumn_typed<binary_blob>> typedColumn = std::dynamic_pointer_cast<icolumn_typed<binary_blob>>( column );
                std::shared_ptr<icell_typed<binary_blob>> cell = typedColumn->createCell(blob);
                row->addCell(cell);
            }
            break;
        default:
            BAJA_ASSERT(false, L"Unsupported data type!");
        }
    }// for each column...
    result->addRow(row);    
}

void db_jet_engine::deleteRow(std::shared_ptr<itable> table,
                              std::shared_ptr<std::vector<std::shared_ptr<icell>>> dummyRow,
                              std::shared_ptr<iresult> dummResult)
{
    jet_call(
        ::JetDelete(
            _sessionId,
            table->getId()
            ));
}

void db_jet_engine::makeKey(std::shared_ptr<itable> table,
                            std::shared_ptr<iindex> index,
                            std::shared_ptr<std::vector<std::shared_ptr<icell>>> criteriaRow,
                            baja::uint32 criteriaCount,
                            key_type keyType)
{
    bool firstIndexColumn = true;
    baja::uint32 indexColumnIndex = 0;
    std::shared_ptr<std::vector<std::shared_ptr<iindex_column>>> indexColumns = index->getIndexColumns();
    std::vector<std::shared_ptr<iindex_column>>::iterator indexColumnIt;
    for (indexColumnIt = indexColumns->begin(); (indexColumnIt < indexColumns->end()) && (indexColumnIndex < criteriaCount); ++indexColumnIt)
    {
        bool lastIndexColumn = ((indexColumnIt + 1) == indexColumns->end()) || (indexColumnIndex + 1 == criteriaCount);
        std::shared_ptr<iindex_column> indexColumn = *indexColumnIt;

        std::vector<std::shared_ptr<icell>>::iterator cellIt;
        for (cellIt = criteriaRow->begin(); cellIt < criteriaRow->end(); ++ cellIt)
        {
            std::shared_ptr<icell> cell = *cellIt;
            if ( cell->getColumnId() == indexColumn->getColumn()->getId() )
            {
                JET_GRBIT flags = 0;
                if (firstIndexColumn == true)
                {
                    flags |= JET_bitNewKey;
                    firstIndexColumn = false;
                }

                if (lastIndexColumn == true)
                {
                    switch (keyType)
                    {
                    case key_type::start_key: flags |= JET_bitFullColumnStartLimit;
                        break;
                    case key_type::end_key: flags |= JET_bitFullColumnEndLimit;
                        break;
                    }
                }

                column_type type = cell->getColumn()->getType();
                switch (type)
                {
                case type_int16:
                    {
                        std::shared_ptr<icell_typed<unsigned short>> typedCell = std::dynamic_pointer_cast<icell_typed<unsigned short>>(cell);
                        unsigned short id = typedCell->getValue();
                        jet_call(
                            ::JetMakeKey(
                                _sessionId,
                                table->getId(),
                                &id,
                                sizeof(id),
                                flags
                                ));
                    }
                    break;
                case type_int32:
                    {
                        std::shared_ptr<icell_typed<unsigned int>> typedCell = std::dynamic_pointer_cast<icell_typed<unsigned int>>(cell);
                        unsigned int id = typedCell->getValue();
                        jet_call(
                            ::JetMakeKey(
                                _sessionId,
                                table->getId(),
                                &id, sizeof(id),
                                flags
                                ));
                    }
                    break;
                case type_int64:
                    {
                        std::shared_ptr<icell_typed<unsigned __int64>> typedCell = std::dynamic_pointer_cast<icell_typed<unsigned __int64>>(cell);
                        unsigned __int64 id = typedCell->getValue();
                        jet_call(
                            ::JetMakeKey(
                                _sessionId,
                                table->getId(),
                                &id,
                                sizeof(id),
                                flags
                                ));
                    }
                    break;
                case type_float32:
                    {
                        std::shared_ptr<icell_typed<float>> typedCell = std::dynamic_pointer_cast<icell_typed<float>>(cell);
                        float id = typedCell->getValue();
                        jet_call(
                            ::JetMakeKey(
                                _sessionId,
                                table->getId(),
                                &id,
                                sizeof(id),
                                flags
                                ));
                    }
                    break;
                case type_string:
                case type_string_long:
                    {
                        std::shared_ptr<icell_typed<std::wstring>> typedCell = std::dynamic_pointer_cast<icell_typed<std::wstring>>(cell);
                        std::wstring id = typedCell->getValue();
                        jet_call(
                            ::JetMakeKey(
                                _sessionId,
                                table->getId(),
                                id.c_str(),
                                static_cast<unsigned long>(id.size() * sizeof(wchar_t)),
                                flags
                                ));
                    }
                    break;
                default:
                    BAJA_THROW(std::logic_error("type cannot be used as a key!"));
                }
                // done with this column.
                break;
            }
        }
        ++indexColumnIndex;
    }
}

bool db_jet_engine::setStartKey(std::shared_ptr<itable> table,
                                std::shared_ptr<iindex> index,
                                std::shared_ptr<std::vector<std::shared_ptr<icell>>> criteriaRow,
                                baja::uint32 criteriaCount,
                                selection_type selectionType)
{
    JET_ERR err = JET_errSuccess;

    makeKey(table,
            index,
            criteriaRow,
            criteriaCount,
            key_type::start_key);

    switch (selectionType)
    {
    case select_equal:
        err = ::JetSeek(_sessionId, table->getId(), JET_bitSeekEQ );
        break;
    case select_equal_or_great_than:
        err = ::JetSeek(_sessionId, table->getId(), JET_bitSeekGE );
        break;
    }

    return err >= 0;
}

bool db_jet_engine::setEndKey(std::shared_ptr<itable> table,
                              std::shared_ptr<iindex> index,
                              std::shared_ptr<std::vector<std::shared_ptr<icell>>> criteriaRow,
                              baja::uint32 criteriaCount)
{
    JET_ERR err = JET_errSuccess;

    makeKey(table,
            index,
            criteriaRow,
            criteriaCount,
            key_type::end_key);

    err = ::JetSetIndexRange(_sessionId, table->getId(), JET_bitRangeInclusive | JET_bitRangeUpperLimit);
    return err >= 0;
}

void db_jet_engine::forEachMatch(std::shared_ptr<itable> table,
                                 std::shared_ptr<iindex> index,
                                 std::shared_ptr<std::vector<std::shared_ptr<icell>>> criteriaRow,
                                 baja::uint32 criteriaCount,
                                 bool assumeSingleMatch,
                                 for_each_func func,
                                 std::shared_ptr<iresult> result)
{
    jet_call(
        ::JetSetCurrentIndex4(
            _sessionId,
            table->getId(),
            index->getName().c_str(),
            0,  // pindexid
            0,  // grbit
            1   // itagSequence
            ));

    selection_type selectionType = assumeSingleMatch ? selection_type::select_equal : selection_type::select_equal_or_great_than;

    if (false == setStartKey(table, index, criteriaRow, criteriaCount, selectionType))
    {
        return;
    }

    if (false == setEndKey(table, index, criteriaRow, criteriaCount))
    {
        return;
    }

    // Loop from start key to end key...
    JET_ERR err = JET_errSuccess;
    do
    {
        func(table, criteriaRow, result);

        err = ::JetMove(_sessionId, table->getId(), JET_MoveNext, 0);
    } while (!(err < 0));
}

void db_jet_engine::forAll(std::shared_ptr<itable> table,
                           std::shared_ptr<iindex> index,
                           for_each_func func,
                           std::shared_ptr<iresult> result)
{
    JET_ERR err = JET_errSuccess;

    jet_call(
        ::JetSetCurrentIndex4(
            _sessionId,
            table->getId(),
            index->getName().c_str(),
            0,  // pindexid
            0,  // grbit
            1   // itagSequence
            ));

    for(err = ::JetMove(_sessionId, table->getId(), JET_MoveFirst, 0);
        JET_errSuccess == err;
        err = ::JetMove(_sessionId, table->getId(), JET_MoveNext, 0))
    {
        func(table, nullptr, result);
    }
}

std::shared_ptr<iengine> baja::storage::createEngine()
{
    return db_jet_engine::create();
}
